package com.fine.spark.connector.base.beans

import com.fine.spark.connector.base.enums.TagEnum
import com.fine.spark.connector.base.utils.Logging
import org.apache.commons.lang3.StringUtils

import java.lang.reflect.Field
import java.util
import scala.beans.BeanProperty

/**
 * @author jayce
 * @version 1.0
 * @date 2021/11/20 3:41 PM
 */
trait BaseConfig extends Logging with Serializable {

  /**
   * 用来定义处理的阶段
   */
  @BeanProperty
  var tag: String = TagEnum.ROOT.toString

  /**
   * 每个处理阶段的名字
   */
  @BeanProperty
  var `type`: String = _

  /**
   * 是否缓存
   */
  @BeanProperty
  var cache: Boolean = false;

  @BeanProperty
  var workerClass: String = _

  def check(): Unit = {
    typeCheck()
  }

  def typeCheck(): Unit = {
    require(StringUtils.isNotBlank(this.`type`),
      s"In node '${this.`type`}', 'type' is required in item.")
  }

  def checkNotNull(fieldNames: String*): Unit = {
    val fields: Map[String, Field] = zipField2Values()
    val nullFields: Seq[String] = fieldNames.filter((fieldName: String) => {
      val field: Field = getValueByFieldName(fields, fieldName)
      field.setAccessible(true)
      val value: AnyRef = field.get(this)
      if (field.getType.isAssignableFrom(classOf[util.Collection[Object]])) {
        val objects: util.Collection[Object] = value.asInstanceOf[util.Collection[Object]]
        objects.isEmpty
      } else {
        value == null
      }
    })
    if (nullFields.nonEmpty) {
      throw new IllegalArgumentException(s"In node '${this.tag}', '${nullFields.mkString(",")}' is must not null!")
    }
  }

  def checkNotBlank(fieldNames: String*): Unit = {
    val fields: Map[String, Field] = zipField2Values()
    val blankFields: Seq[String] = fieldNames.filter((fieldName: String) => {
      val field: Field = getValueByFieldName(fields, fieldName)
      field.setAccessible(true)
      val value: AnyRef = field.get(this)
      if (field.getType.isAssignableFrom(classOf[String])) {
        StringUtils.isBlank(value.toString)
      } else {
        value == null
      }
    })
    if (blankFields.nonEmpty) {
      throw new IllegalArgumentException(s"In node '${this.tag}', '${blankFields.mkString(",")}' is must not blank!")
    }
  }

  /**
   * 根据字段名获取字段的值值
   *
   * @param fields    fields
   * @param fieldName fieldName
   * @return
   */
  private def getValueByFieldName(fields: Map[String, Field], fieldName: String): Field = {
    val field = fields.getOrElse(fieldName, null)
    require(field != null, s"In node '${this.tag}, checked field '$fieldName'," +
      s" this field isn't a member variable in config bean '${this.getClass.getName}'," +
      s" contact the developer!")
    field
  }

  /**
   * 获取当前类的所有属性和值
   *
   * @return
   */
  private def zipField2Values(): Map[String, Field] = {
    typeCheck()
    val fields = scala.collection.mutable.HashMap.empty[String, Field]
    var classObj: Class[_ <: Any] = this.getClass
    while (classObj.getName != classOf[Object].getName) {
      classObj.getDeclaredFields.map((field: Field) => {
        fields += (field.getName -> field)
      })
      classObj = classObj.getSuperclass()
    }
    fields.toMap
  }
}
